/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.polarrose.spring.bayeux;

import com.polarrose.json.JsonSerializer;
import com.polarrose.spring.bayeux.util.BayeuxUtil;
import com.polarrose.spring.bayeux.util.JsonUtil;
import org.mortbay.cometd.Channel;
import org.mortbay.cometd.Client;
import org.mortbay.cometd.DataFilter;
import org.mortbay.util.ajax.JSON;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Required;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author Stefan Arentz
 */

public class BayeuxChannel implements InitializingBean
{
    /**
     *
     */

    private BayeuxController bayeuxController;

    @Required
    public void setBayeuxController(BayeuxController bayeuxController)
    {
        this.bayeuxController = bayeuxController;
    }

    /**
     *
     */

    private String name;

    @Required
    public void setName(String name)
    {
        this.name = name;
    }

    public String getName()
    {
        return name;
    }

    /**
     * 
     */

    private Channel channel;

    Channel getChannel()
    {
        return channel;
    }

    /**
     * 
     */

    private List<BayeuxFilter> filters = new ArrayList<BayeuxFilter>();

    public void setFilters(List<BayeuxFilter> filters)
    {
        this.filters = filters;
    }

    //

    public void publish(Object object)
    {
        try {
            JsonSerializer serializer = new JsonSerializer();
            String json = serializer.serialize(object);
            bayeuxController.getBayeux().publish(channel.getId(), null, JSON.parse(json), "someMessageId"); // TODO messageId?
        } catch (IOException e) {
            // TODO Ignored for now
        }
    }

    public void publish(Object object, BayeuxClient from)
    {
        bayeuxController.getBayeux().publish(
            channel.getId(),
             from.getClient(),
            JsonUtil.transmogrify(object),
            "someMessageId"  // TODO messageId?
        );
    }

    public void addSubscriber(BayeuxClient client)
    {
        channel.addSubscriber(client.getClient());
    }

    public BayeuxChannel getReplyChannel(BayeuxClient client)
    {
        Channel channel = bayeuxController.getBayeux().getChannel(getName() + "/" + client.getId(), false);
        if (channel != null) {
            return new BayeuxChannel(bayeuxController, getName() + "/" + client.getId());
        } else {
            return null;
        }
    }

    public BayeuxDestination getReplyDestination(BayeuxClient client)
    {
        return new BayeuxDestination(getName() + "/" + client.getId());
    }

    //

    public BayeuxChannel(BayeuxController bayeuxController, String name)
    {
        this.bayeuxController = bayeuxController;
        this.name = name;

        channel = bayeuxController.getBayeux().getChannel(name);
        if (channel == null) {
            channel = bayeuxController.getBayeux().getChannel(name, true);
        }
    }

    public BayeuxChannel()
    {
    }

    /**
     * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
     */

    public void afterPropertiesSet() throws Exception
    {
        channel = bayeuxController.getBayeux().getChannel(name);
        if (channel == null) {
            channel = bayeuxController.getBayeux().getChannel(name, true);
        }

        for (BayeuxFilter filter : filters) {
            channel.addDataFilter(new DataFilterAdapter(filter));
        }
    }

    //

    class DataFilterAdapter implements DataFilter
    {
        private final Class<?> filterDataClass;
        private final BayeuxFilter bayeuxFilter;

        public DataFilterAdapter(BayeuxFilter bayeuxFilter)
        {
            this.filterDataClass = BayeuxUtil.getBayeuxFilterDataClass(bayeuxFilter);
            this.bayeuxFilter = bayeuxFilter;
        }

        public void init(Object init)
        {
            // TODO What it this?
        }

        public Object filter(Client from, Channel to, Object data) throws IllegalStateException
        {
            if (filterDataClass != null) {
                return JsonUtil.transmogrify(bayeuxFilter.filter(bayeuxController.getBayeux(), from, JsonUtil.transmogrify(filterDataClass, data, null)));
            } else {
                return bayeuxFilter.filter(bayeuxController.getBayeux(), from, data);
            }
        }
    }

    //

    public String toString()
    {
        return "BayeuxChannelImpl[name=" + getName() + "]";
    }
}
